package com.eian.boot.redis.queue;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 初始化Redis队列监听器
 *
 * @author alex.meng
 * @createTime 2024-09-09 22:49
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisQueueInit implements ApplicationContextAware {
    public static final String REDIS_QUEUE_PREFIX = "redis-queue";

    final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
    private final RedissonClient redisson;
    private ExecutorService executorService;

    public static String buildQueueName(String queueName) {
        return REDIS_QUEUE_PREFIX + ":" + queueName;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, RedisQueueListener> queueListeners = applicationContext.getBeansOfType(RedisQueueListener.class);
        if (!queueListeners.isEmpty()) {
            executorService = createThreadPool();
            for (Map.Entry<String, RedisQueueListener> entry : queueListeners.entrySet()) {
                RedisQueue redisQueue = entry.getValue().getClass().getAnnotation(RedisQueue.class);
                if (redisQueue != null) {
                    String queueName = redisQueue.value();
                    executorService.submit(() -> listenQueue(queueName, entry.getValue()));
                }
            }
        }
    }

    private ExecutorService createThreadPool() {
        return new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors() * 2,
                Runtime.getRuntime().availableProcessors() * 4,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new NamedThreadFactory(REDIS_QUEUE_PREFIX),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    private void listenQueue(String queueName, RedisQueueListener redisQueueListener) {
        queueName = buildQueueName(queueName);
        RBlockingQueue<?> blockingQueue = redisson.getBlockingQueue(queueName);
        log.info("Redis队列监听开启: {}", queueName);
        while (!shutdownRequested.get() && !redisson.isShutdown()) {
            try {
                Object message = blockingQueue.take();
                executorService.submit(() -> redisQueueListener.consume(message));
            } catch (RedissonShutdownException e) {
                log.info("Redis连接关闭，停止监听队列: {}", queueName);
                break;
            } catch (Exception e) {
                log.error("监听队列异常: {}", queueName, e);
            }
        }
    }

    public void shutdown() {
        if (executorService != null) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException ex) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        shutdownRequested.set(true);
        if (redisson != null && !redisson.isShuttingDown()) {
            redisson.shutdown();
        }
    }

    private static class NamedThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        public NamedThreadFactory(String prefix) {
            this.namePrefix = prefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, namePrefix + "-" + threadNumber.getAndIncrement());
        }
    }

}